#include "gateway_server.h"


//////////////////////////////////////////////////////////////////////////
c2s_network_handler::c2s_network_handler(gateway_server *pserver)
{
    m_pserver = pserver;
}
void c2s_network_handler::handle_accepted(session<c2s_network_handler> &session_)
{
    DEBUG4("c2s accepted connection:", session_.get_connection_index(),":",session_.get_unique_key());
    per_head newconnection;
    newconnection.connection_id = session_.get_connection_index();
    newconnection.connection_key = session_.get_unique_key();
    newconnection.pkg_size = CONN_ESTA;
    m_pserver->m_circular_buffer.push(&newconnection, sizeof(newconnection));
}
void c2s_network_handler::handle_read(session<c2s_network_handler> &session_, const char *buffer_, std::size_t &bytes_transferred)
{
    DEBUG6("c2s connection: ",session_.get_connection_index(),":",session_.get_unique_key()," read:", bytes_transferred);
    per_head connectin_pkg;
    connectin_pkg.connection_id = session_.get_connection_index();
    connectin_pkg.connection_key = session_.get_unique_key();
    connectin_pkg.pkg_size = (int)bytes_transferred;    

    m_pserver->m_circular_buffer.thread_lock();
    m_pserver->m_circular_buffer.push(&connectin_pkg, sizeof(connectin_pkg));
    m_pserver->m_circular_buffer.push(buffer_, (int)bytes_transferred);
    m_pserver->m_circular_buffer.thread_unlock();
}
void c2s_network_handler::handle_close(session<c2s_network_handler> &session_)
{
    DEBUG5("c2s connection:", session_.get_connection_index(),":",session_.get_unique_key()," closed");
    per_head connectionlost;
    connectionlost.connection_id = session_.get_connection_index();
    connectionlost.connection_key = session_.get_unique_key();
    connectionlost.pkg_size = CONN_LOST;
    m_pserver->m_circular_buffer.push(&connectionlost, sizeof(connectionlost));
}


//////////////////////////////////////////////////////////////////////////
s2s_network_handler::s2s_network_handler(gateway_server *pserver)
{
    m_pserver = pserver;
}

void s2s_network_handler::handle_accepted(session<s2s_network_handler> &session_)
{
    DEBUG5("s2s connection:", session_.get_connection_index(),":",session_.get_unique_key()," accepted");    
    m_pserver->m_ps2s_thread->post_msg(new s2s_thread::set_session(session_));
}
void s2s_network_handler::handle_read(session<s2s_network_handler> &session_, const char *buffer_, std::size_t &bytes_transferred)
{
    DEBUG6("s2s connection: ",session_.get_connection_index(),":",session_.get_unique_key()," read:", bytes_transferred);
}
void s2s_network_handler::handle_close(session<s2s_network_handler> &session_)
{
    DEBUG5("s2s connection:", session_.get_connection_index(),":",session_.get_unique_key()," closed");
    m_pserver->m_ps2s_thread->post_msg(new s2s_thread::session_closed(session_));
}

//////////////////////////////////////////////////////////////////////////
s2s_thread::s2s_thread(int gatewayid,
                       circular_buffer &the_circular_buffer):m_gateway_id(gatewayid),m_circular_buffer(the_circular_buffer),
            threadbase<s2s_thread>(&s2s_thread::thread_run)
{
    m_psession = NULL;
    m_send_buffer = new char[NET_TRAFFIC_PER_SEC];
    m_send_size = 0;
    m_serial_no = 0;
    set_runner(this);
}

s2s_thread::~s2s_thread()
{
    if (m_psession)
    {
        delete m_psession;
    }
    if (m_send_buffer)
    {
        delete [] m_send_buffer;
    }
}

void s2s_thread::thread_run()
{
    m_thread_mutex.enter();
    for (std::list<thread_msg *>::iterator itor=m_thread_msgs.begin(); itor!=m_thread_msgs.end(); ++itor)
    {
        if ((*itor)->type==SET_SESSION)
        {
            set_session *pmsg = (set_session *)(*itor);
            if (m_psession==NULL)
            {
                m_psession = new session<s2s_network_handler>(pmsg->m_session);
                DEBUG4("s2s session setted, session ",m_psession->get_connection_index(),":",m_psession->get_unique_key());
            }
            else
            {
                DEBUG4("s2s session already exist , session ",m_psession->get_connection_index(),":",m_psession->get_unique_key());
            }
        }
        else if ((*itor)->type==SESSION_CLOSED)
        {
            session_closed *pmsg = (session_closed *)(*itor);
            if (m_psession==NULL)
            {
                DEBUG5("the closed session ",m_psession->get_connection_index(),":",m_psession->get_unique_key()," is not current session null");
            }
            else
            {
                if (*m_psession==pmsg->m_session)
                {
                    DEBUG4("s2s session closed , session ",m_psession->get_connection_index(),":",m_psession->get_unique_key());
                    delete m_psession;
                    m_psession = NULL;                    
                }
                else
                {
                    DEBUG8("the closed session ",m_psession->get_connection_index(),":",m_psession->get_unique_key()," is not current session ",m_psession->get_connection_index(),":",m_psession->get_unique_key());
                }
            }
        }
        delete (*itor);
    }
    m_thread_msgs.clear();
    m_thread_mutex.leave();
    
    if (m_send_size<=0)
    {
        int getted_size = m_circular_buffer.get_copy(m_send_buffer+sizeof(total_head), NET_TRAFFIC_PER_SEC-sizeof(total_head));
        if (getted_size>0)
        {
            total_head *phead = (total_head*)m_send_buffer;
            phead->serial_no = m_serial_no;
            phead->gateway_id = m_gateway_id;
            phead->pkg_size = getted_size+sizeof(total_head);
            if (m_psession)
            {
                if (m_psession->write(m_send_buffer, phead->pkg_size))
                {
                    DEBUG2("send logic pack success, serial no:", m_serial_no);
                    ++m_serial_no ;
                }
                else
                {
                    DEBUG2("send logic pack failed , try later, serial no:", m_serial_no);
                    m_send_size = phead->pkg_size;
                }        
            }
        }        
    }
    else
    {
        if (m_psession)
        {
            if (m_psession->write(m_send_buffer, m_send_size))
            {
                DEBUG2("send logic pack success, serial no:", m_serial_no);
                ++m_serial_no;
                m_send_size = 0;
            }
            else
            {
                DEBUG2("send logic pack failed , try later, serial no:", m_serial_no);                
            }
        }
    }
    
    Sleep(1000);    // once a second
}

const char *s2s_thread::thread_name()
{
    return "s2s_thread";
}

void s2s_thread::post_msg(thread_msg *pmsg)
{
    recursive_mutex::scoped_lock lock(m_thread_mutex);
    m_thread_msgs.push_back(pmsg);
}


//////////////////////////////////////////////////////////////////////////
gateway_server::gateway_server()
{
    m_port_c2s = 8888;
    m_port_s2s = 8889;

    m_ps2s_thread = new s2s_thread(0, m_circular_buffer);
}

gateway_server::~gateway_server()
{
    if (m_ps2s_thread)
    {
        delete m_ps2s_thread;
    }
}

void gateway_server::start_server()
{
    boost::shared_ptr<c2s_network_handler> pc2shandler(new c2s_network_handler(this));
    boost::shared_ptr<s2s_network_handler> ps2shandler(new s2s_network_handler(this));

    m_client_to_server = m_server_manager.start_an_server<c2s_network_handler>("0.0.0.0", lexical_cast<std::string,int>(m_port_c2s), pc2shandler);
    m_server_to_server = m_server_manager.start_an_server<s2s_network_handler>("0.0.0.0", lexical_cast<std::string,int>(m_port_s2s), ps2shandler);

    m_ps2s_thread->thread_begin();
}

void gateway_server::stop_server()
{
    m_client_to_server.reset();
    m_server_to_server.reset();

    m_ps2s_thread->thread_syncstop();
}

